package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @author Felix
 * @date 2023/2/10
 * 查询维度的工具类
 */
public class DimUtil {
    public static JSONObject getDimInfo(String tableName,String id){
        return getDimInfo(tableName,Tuple2.of("id",id));
    }
    /**
     * 查询维度数据：加入了旁路缓存优化
     * 旁路缓存思路：先从缓存中获取维度数据，如果缓存中能够找到对应的维度，那么将其作为方法的返回值进行返回(缓存命中)；
     *          如果从缓存中没有找到对应的维度数据，那么发送请求到Phoenix表中去查询维度数据，并将查询的结果放到缓存
     *          中缓存起来，方便下次查询使用。
     * 缓存产品的选择
     *      状态：  性能更好，维护性差
     *      Redis：性能也不错，维护性好   √
     * Redis中的相关配置
     *      类型：String
     *      key: dim:维度表表名:主键值1_主键值2   dim:dim_base_trademark:1_三星
     *      expire: 1day   避免冷数据常驻内存
     *      注意：如果Phoenix表中的维度数据发生了变化，需要将缓存的数据清除掉，避免数据出现一致性问题
     * */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues){
        //拼接从Redis中查询数据的key
        StringBuilder redisKey = new StringBuilder("dim:"+tableName.toLowerCase()+":");

        //拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where ");

        //遍历可变长参数，获取每一对查询条件
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + " ='" + columnValue + "' ");
            redisKey.append(columnValue);
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        Jedis jedis = null;
        String dimInfoJsonStr = null;
        JSONObject dimJsonObj = null;

        try {
            // 先从缓存中获取维度数据
            jedis = RedisUtil.getJedis();
            dimInfoJsonStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            System.out.println("从Redis中查询维度数据发生了异常");
        }

        //判断是否从Redis缓存获取到了维度数据
        if(StringUtils.isNotEmpty(dimInfoJsonStr)){
            // 如果缓存中能够找到对应的维度，那么将其作为方法的返回值进行返回(缓存命中)
            dimJsonObj = JSON.parseObject(dimInfoJsonStr);
        }else {
            // 如果从缓存中没有找到对应的维度数据，那么发送请求到Phoenix表中去查询维度数据
            System.out.println("从Phoenix表中查询维度的SQL：" + selectSql);
            //底层还是调用的是PhoenixUtil.queryList方法
            List<JSONObject> dimJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            if (dimJsonObjList != null && dimJsonObjList.size() > 0) {
                //注意：因为我们这里是根据维度的主键，获取维度数据，所以集合中如果有元素的话，只会有一条
                dimJsonObj = dimJsonObjList.get(0);
                // 将查询的结果放到缓存中缓存起来，方便下次查询使用
                if(jedis != null){
                    jedis.setex(redisKey.toString(),24*60*60,dimJsonObj.toJSONString());
                }
            } else {
                System.out.println("没有找到对应的维度数据");
            }
        }
        if(jedis != null){
            System.out.println("~~~关闭Jedis客户端~~~");
            jedis.close();
        }
        return dimJsonObj;
    }
    //查询维度数据  没有使用任何优化方案
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where ");
        //遍历可变长参数，获取每一对查询条件
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + " ='" + columnValue + "' ");
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
            }
        }
        System.out.println("从Phoenix表中查询维度的SQL：" + selectSql);
        //底层还是调用的是PhoenixUtil.queryList方法
        List<JSONObject> dimJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject dimJsonObj = null;
        if (dimJsonObjList != null && dimJsonObjList.size() > 0) {
            //注意：因为我们这里是根据维度的主键，获取维度数据，所以集合中如果有元素的话，只会有一条
            dimJsonObj = dimJsonObjList.get(0);
        } else {
            System.out.println("没有找到对应的维度数据");
        }
        return dimJsonObj;
    }

    public static void main(String[] args) {
        // System.out.println(getDimInfoNoCache("dim_base_trademark" , Tuple2.of("id","1")));
        // System.out.println(getDimInfo("dim_base_trademark" , Tuple2.of("id","1")));
        System.out.println(getDimInfo("dim_base_trademark" , "1"));
    }

    public static void delCached(String tableName,String id){
        String redisKey = "dim:"+tableName.toLowerCase()+":"+id;
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("从Redis中删除缓存的维度数据发生了异常");
        } finally {
            if(jedis != null){
                System.out.println("~~~删除缓存数据后，关闭Jedis客户端~~~");
                jedis.close();
            }
        }
    }
}
